package com.zhengb.flink.study.log.analysis.watermark;

import com.zhengb.flink.study.log.analysis.entity.LogMessageSource;
import com.zhengbo.flink.demo.base.utils.DateUtil;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

/**
 * @author zhengbo
 * @date 2020/2/6
 */
public class TimeLagWatermarkGenerator implements AssignerWithPeriodicWatermarks<LogMessageSource> {

    private final static long MAX_TIME_TAG = 5000L;

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(System.currentTimeMillis() - MAX_TIME_TAG);
    }

    @Override
    public long extractTimestamp(LogMessageSource element, long previousElementTimestamp) {

        return DateUtil.parseDateTime(element.getDataTime()).getTime();
    }
}
